#include <amqpcpp.h>
#include <amqpcpp/libboostasio.h>

#include "core.h"
#include "server_impl.h"
#include <string>
#include <iostream>
#include <stdio.h>
#include <stdlib.h>

#include <boost/algorithm/string.hpp>  
#include <boost/lexical_cast.hpp> 


#include "libzqutil/log.h"
#include "client_impl.h"
#include "error.h"

namespace libzq
{

	ServerImpl::ServerImpl()
	{

	}

	ServerImpl::~ServerImpl()
	{
	}

	void ServerImpl::start_serve(const char* host, uint16_t port)
	{
		CORE.install_stop([this]()
		{
			stop();
		});
		tcp::endpoint addr(address::from_string(host), port);

		_stop = false;
		go[this, &addr]()
		{
			bool stop;
			_stop_chan >> stop;
			auto sock = shared_ptr<tcp::socket>(new tcp::socket(CORE.ioservice()));
			try
			{
				sock->connect(addr); // 让accept返回
				sock->shutdown(boost::asio::ip::tcp::socket::shutdown_both);
				sock->close();
			}
			catch (boost::system::error_code& ec)
			{

			}
		};

		_acceptor = shared_ptr<tcp::acceptor>(new tcp::acceptor(CORE.ioservice(), addr, true));
		LOGI("Listen on %s:%d", host, port);
		while (1)
		{
			auto sock = shared_ptr<tcp::socket>(new tcp::socket(CORE.ioservice()));
			try
			{
				_acceptor->accept(*sock);
			}
			catch (boost::system::error_code& ec)
			{
				LOGE("Accept error [%s]", ec.message());
				continue;
			}
			if (_stop)
			{
				LOGT("Stop accept new connection");
				break;
			}
			connection_connected(sock);
		};
		try
		{
			_acceptor->close();
		}
		catch (boost::system::error_code& ec)
		{
			LOGE("Close acceptor error [%s]", ec.message());
		}
	}

	void ServerImpl::stop()
	{
		LOGI("Start close server");
		_stop = true;
		_stop_chan << true;
		for (auto itor = _clients.begin(); itor != _clients.end(); ++itor)
		{
			(*itor)->stop(ERROR_STOP_SERVICE);
		}
		LOGI("Close connection");
		_clients.clear();
	}


	void ServerImpl::connection_connected(shared_ptr<tcp::socket> sock)
	{
		try
		{
			auto endpoint = sock->remote_endpoint();
			LOGT("New Client from [%s:%d] connected", endpoint.address().to_string(), endpoint.port());
		}
		catch (boost::system::error_code& ec)
		{
			LOGT("New Client connected");
		}
		shared_ptr<ClientImpl> client = make_shared<ClientImpl>(sock);
		_clients.push_back(client);
		client->set_disconnect_cb(bind(&ServerImpl::connection_disconnect, this, std::placeholders::_1));
		client->start_serve();
	}

	void ServerImpl::connection_disconnect(shared_ptr<ClientImpl> client)
	{
		LOGT("Connection disconnected");
		if (!_stop) _clients.erase(std::remove(_clients.begin(), _clients.end(), client), _clients.end());

	}
	void ServerImpl::boradcast(const char* host, const char* queue, const char* exchange, const string& exchange_type, const char* routing_key)
	{
		static AMQP::LibBoostAsioHandler handler(CORE.ioservice());
		static AMQP::TcpConnection connection(&handler, AMQP::Address(host));
		if (exchange_type.compare("fanout") && exchange_type.compare("topic"))
		{
			LOGE("Rabbitmq exchange error with [ %s ]", exchange_type);
			return;
		}
		auto type = !exchange_type.compare("fanout") ? AMQP::fanout : AMQP::topic;
		auto channel = std::make_shared<AMQP::TcpChannel>(&connection);

		CORE.install_stop([channel]()
		{
			channel->close();
		});
		auto on_consume = [this, channel, queue](const AMQP::Message &message, uint64_t deliveryTag, bool redelivered)
		{
			LOGT("Consumed from queue [ %s ] exchange [ %s ] routing key [ %s ], len [ %d ]", queue, message.exchange(), message.routingkey(), message.bodySize());

			std::shared_ptr<MsgPacker> msg_ptr = MsgPacker::create(message.body(), message.bodySize());
			if (msg_ptr)
			{
				go std::bind(&ServerImpl::on_boradcast, this, msg_ptr);
			}
			else
			{
				LOGW("Error boradcast msg");
			}
			channel->ack(deliveryTag);
		};

		auto consume = [on_consume, queue, channel]()
		{
			channel->consume(queue).onReceived(on_consume);
		};

		auto bind_queue = [consume, channel, queue, exchange, routing_key]()
		{
			channel->bindQueue(exchange, queue, routing_key).onSuccess(consume);
			LOGI("Declared exchange [ %s ]", exchange);
		};

		auto dealare_exchange = [bind_queue, channel, exchange, type](const std::string &name, uint32_t messagecount, uint32_t consumercount)
		{
			LOGI("Declared queue [ %s ]", name);
			channel->declareExchange(exchange, type).onSuccess(bind_queue);
		};
		channel->declareQueue(queue).onSuccess(dealare_exchange);
	}
	void ServerImpl::on_boradcast(std::shared_ptr<MsgPacker> msg)
	{
		LOGT("Recv boradcast msg, msgid [ %s ]", msg->msgid());
	}
}